package com.rr.streamrabbitmqnew.service;

//import com.banggood.sp.autoconfigure.stream.common.CommonEventReq;
//import com.banggood.sp.autoconfigure.stream.common.message.ListerPublishBody;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.util.function.Tuple2;

import java.util.function.Consumer;
import java.util.function.Function;

/**
 * @author zhuangcanrun
 * @Date 2021/06/17/9:47
 * @description
 **/

@Configuration
public class ReceiveService {

    @Value("spring.cloud.stream.bindings.consumer1-in-0.group")
    private String consumer1Group;

    //接收的消息也可以用一个Message<>，然后可以拿到请求头信息那些
    @Bean
    public Consumer<Message<String>> consumer1(){
        return str -> {
            String payload = str.getPayload();
            MessageHeaders headers = str.getHeaders();
            System.out.println("consumer1接收到的消息:{" + str + "}");
        };
    }

    @Bean
    public Consumer<String> consumer2(){
        return str -> {
            System.out.println("consumer2接收到的消息:{" + str + "}");
        };
    }

    @Bean
    public Function<Message<String>, Message<String>> enrich() {
        return message -> {
            Assert.isTrue(!message.getHeaders().containsKey("foo"), "Should NOT contain 'foo' header");
            return MessageBuilder.fromMessage(message).setHeader("foo", "bar").build();
        };
    }

    @Bean
    public Function<Message<String>, Message<String>> echo() {
        return message -> {
            Assert.isTrue(message.getHeaders().containsKey("foo"), "Should contain 'foo' header");
            System.out.println("Incoming message " + message);
            return message;
        };
    }

    @Bean
    public Function<Tuple2<Flux<String>,Flux<Integer>>,Flux<String>> gather(){
        return tuple -> {
            Flux<String> stringStream = tuple.getT1();
            Flux<String> intStream = tuple.getT2().map(i -> String.valueOf(i));
            return Flux.merge(stringStream, intStream);
        };
    }

    @Bean
    public Consumer<String> even() {
        return value -> System.out.println("EVEN: " + value);
    }

    @Bean
    public Consumer<String> odd() {
        return value -> System.out.println("ODD: " + value);
    }

//    @Bean
//    public Consumer<Message<CommonEventReq<ListerPublishBody>>> listerPublish(){
//        return msg -> {
//            CommonEventReq<ListerPublishBody> req = msg.getPayload();
//            ListerPublishBody body = req.getBody();
//            System.out.println(req.getPlatformNo());
//        };
//    }


//    @Bean
//    public Consumer<String> stringSupplier(){
//        return str->{
//            System.out.println(str);
//        };
//    }

}
